package com.db;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class HBaseMain {

  private static final String ZK_POS = "47.101.204.23,47.101.216.12,47.101.206.249";
  private static final String ZK_PORTS = "2181";
  private static final String TABLE_NAME = "student";
  private static final String NAMESPACE = "huanggengkeng";

  public static void main(String[] args) throws IOException {

    HBaseUtils hBaseUtils = new HBaseUtils(ZK_POS, ZK_PORTS);

    // check namespace exist
    NamespaceDescriptor namespaceDescriptor = null;
    if (!hBaseUtils.existsNamespace(NAMESPACE)) {
      namespaceDescriptor = hBaseUtils.createNamespace(NAMESPACE);
      // 修改namespace的配置属性
      hBaseUtils.modifyNamespaceConfig(namespaceDescriptor, "PROPERTY_NAME", "PROPERTY_VALUE");
      System.out.println("创建namespace:" + NAMESPACE + ",成功！");
    }

    // 查看该namespace的详情
    System.out.println("查看namespace的详情===>" + hBaseUtils.getNamespace(NAMESPACE) + "\n");

    if (hBaseUtils.existTable(NAMESPACE + ":" + TABLE_NAME)) {
      hBaseUtils.deleteTable(NAMESPACE + ":" + TABLE_NAME);
    }
    // 创建表 并声明命名空间
    hBaseUtils.createTable(NAMESPACE + ":" + TABLE_NAME, "info", "score");

    // 列出default命名空间的表
    List<TableDescriptor> tablesDesc = hBaseUtils.listTableDescriptorsByNamespace(NAMESPACE);
    System.out.println(NAMESPACE + "命名空间下的表有：" + tablesDesc);
    // 插入数据
    insertData(hBaseUtils);
    // 扫描数据
    scanTable(hBaseUtils);
    // 关闭连接
    hBaseUtils.close();
  }

  /** 向student 表插入数据 */
  public static void insertData(HBaseUtils hBaseUtils) throws IOException {
    Table table = hBaseUtils.getTable(TableName.valueOf(NAMESPACE + ":" + TABLE_NAME));
    List<Put> puts = new ArrayList<>();
    byte[] info = "info".getBytes(); // 列簇 -info
    byte[] score = "score".getBytes(); // 列簇  - score
    byte[] id = "student_id".getBytes(); // 列字段 -student_id
    byte[] cla = "class".getBytes(); // 列字段 - class
    byte[] understanding = "understanding".getBytes(); // 列字段 - understanding
    byte[] programming = "programming".getBytes(); // 列字段 - programming
    Put put = new Put("Tom".getBytes());
    put.addColumn(info, id, "20210000001".getBytes());
    put.addColumn(info, cla, "1".getBytes());
    put.addColumn(score, understanding, "75".getBytes());
    put.addColumn(score, programming, "82".getBytes());
    puts.add(put);
    put = new Put("Tom".getBytes());
    put.addColumn(info, id, "20210000001".getBytes());
    put.addColumn(info, cla, "1".getBytes());
    put.addColumn(score, understanding, "75".getBytes());
    put.addColumn(score, programming, "82".getBytes());
    puts.add(put);
    put = new Put("Jerry".getBytes());
    put.addColumn(info, id, "20210000002".getBytes());
    put.addColumn(info, cla, "1".getBytes());
    put.addColumn(score, understanding, "85".getBytes());
    put.addColumn(score, programming, "67".getBytes());
    puts.add(put);
    put = new Put("Jack".getBytes());
    put.addColumn(info, id, "20210000003".getBytes());
    put.addColumn(info, cla, "2".getBytes());
    put.addColumn(score, understanding, "80".getBytes());
    put.addColumn(score, programming, "80".getBytes());
    puts.add(put);
    put = new Put("Rose".getBytes());
    put.addColumn(info, id, "20210000004".getBytes());
    put.addColumn(info, cla, "2".getBytes());
    put.addColumn(score, understanding, "60".getBytes());
    put.addColumn(score, programming, "61".getBytes());
    puts.add(put);
    put = new Put("huanggengkeng".getBytes());
    put.addColumn(info, id, "20210000005".getBytes());
    put.addColumn(info, cla, "1".getBytes());
    put.addColumn(score, understanding, "85".getBytes());
    put.addColumn(score, programming, "90".getBytes());
    puts.add(put);

    table.put(puts);
    table.close();
  }

  /** scan student 表数据 */
  public static void scanTable(HBaseUtils hBaseUtils) throws IOException {
    Table table = hBaseUtils.getTable(TableName.valueOf(NAMESPACE + ":" + TABLE_NAME));
    Scan scan = new Scan();
    ResultScanner scanner = table.getScanner(scan);
    for (Result result : scanner) {
      String name = new String(result.getRow());
      Cell[] cells = result.rawCells();
      System.out.println("--------------------------表数据如下------------------------");
      for (Cell cell : cells) {
        String family =
            Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
        String qualifier =
            Bytes.toString(
                cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
        String row = Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
        String value1 =
            Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
        System.out.println(row + "," + family + "," + qualifier + "," + value1);
      }
    }
    scanner.close();
    table.close();
  }
}
